package com.lagou.rpc.zookeeper;

import com.lagou.rpc.common.RequestMetrics;
import com.lagou.rpc.zookeeper.config.ZookeeperConfig;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.framework.state.ConnectionStateListener;
import org.apache.curator.retry.RetryNTimes;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.data.Stat;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class ZookeeperClient {
    static final String ZK_ROOT_PATH = "/lg-rpc-center/";
    private static final String ZK_SPLIT = "/";
    private CuratorFramework client;
    //可用于服务调用的服务端信息<服务名称，服务端ip和端口列表>
    public static final Map<String, List<String>> LOAD_BALANCE_MAP = new HashMap<>();
    //监听器列表
    public static final List<NodeChangeListener> listeners = new ArrayList<>();
    //定时上报线程池
    private static final ScheduledExecutorService REPORT_THREAD_POLL = Executors.newScheduledThreadPool(1);

    /**
     * 通过构造器初始化zookeeper客户端
     *
     * @param zookeeperConfig
     */
    public ZookeeperClient(ZookeeperConfig zookeeperConfig) {
        //获取Curator客户端
        CuratorFrameworkFactory.Builder builder = CuratorFrameworkFactory.builder()
                .connectString(ZookeeperConfig.getInstance().getZookeeperAddress())
                .sessionTimeoutMs(5000)
                .connectionTimeoutMs(5000)
                .retryPolicy(new RetryNTimes(1, 1000));
        client = builder.build();
        client.getConnectionStateListenable().addListener(new ConnectionStateListener() {
            @Override
            public void stateChanged(CuratorFramework curatorFramework, ConnectionState connectionState) {
                boolean connected = connectionState.isConnected();
                if (ConnectionState.CONNECTED.equals(connectionState)) {
                    System.out.println("========连接zookeeper成功========");
                }
            }
        });
        //启动Curator客户端
        client.start();
        //如果是服务端每5秒上报响应时间
        boolean consumerSide = ZookeeperConfig.getInstance().isConsumerSide();
        long interval = ZookeeperConfig.getInstance().getInterval();
        //如果是消费端并且上报时间间隔不为0
        if (consumerSide && interval > 0) {
            //执行定时线程
            REPORT_THREAD_POLL.scheduleWithFixedDelay(new Runnable() {
                @Override
                public void run() {
                    //TODO
                    //获取所有需要上报的节点
                    ConcurrentHashMap<String, String> timeMap = RequestMetrics.getInstance().getTimeMap();
                    System.out.println("\n^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^");
                    System.out.println("^^^^^^^^^^^上报节点响应时间信息：" + timeMap);
                    System.out.println("^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\n");
                    //如果为空则无需上报
                    if (timeMap.isEmpty()) {
                        return;
                    }
                    String metricPath = ZK_ROOT_PATH + "metrics";
                    if (!exist(metricPath)) {
                        //创建服务端上报永久父节点：/lg-rpc-center/metrics
                        create(metricPath, false);
                    }

                    //遍历所有需要上报的节点并修改节点信息
                    timeMap.entrySet().stream().forEach(entry -> {
                        String address = entry.getKey();
                        String value = entry.getValue();
                        String[] split = value.split("#");
                        Long lastRequestTime = Long.parseLong(split[0]);
                        Long costTime = Long.parseLong(split[1]);
                        //当前临时节点路径：/lg-rpc-center/metrics/127.0.0.1:8991
                        String nodePath = metricPath + ZK_SPLIT + address;
                        long intervalTime = System.currentTimeMillis() - lastRequestTime;
                        //判断当前时间举例最后一次请求时间是否超过5s，超过则删除zk节点
                        if (intervalTime > 5000 && exist(nodePath)) {
                            delete(nodePath);
                            System.out.println("!!!!!!!!!!!!!!删除zk节点：" + address + " 距离上次请求时间为：" + intervalTime + "毫秒\n");
                        } else {
                            //添加或修改节点信息
                            updateNodeInfo(nodePath, costTime.toString(), true);
                        }
                    });
                }

            }, interval, interval, TimeUnit.SECONDS);
        }


    }

    /**
     * 删除节点
     *
     * @param nodePath
     */
    private void delete(String nodePath) {
        try {
            client.delete().forPath(nodePath);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 添加或修改节点信息
     *
     * @param nodePath  节点路径
     * @param data      节点数据
     * @param ephemeral 是否为临时节点
     */
    private void updateNodeInfo(String nodePath, String data, boolean ephemeral) {
        try {
            if (exist(nodePath)) {
                //节点存在则修改节点数据
                client.setData().forPath(nodePath, data.getBytes("utf-8"));
            } else {
                //如果节点不存在判断创建永久节点还是临时节点
                CreateMode mode = ephemeral ? CreateMode.EPHEMERAL : CreateMode.PERSISTENT;
                client.create().creatingParentsIfNeeded().withMode(mode).forPath(nodePath, data.getBytes("utf-8"));
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 添加监听者
     * @param listener
     */
    public void addListener(NodeChangeListener listener) {
        listeners.add(listener);
    }

    /**
     * 注册节点
     * @param service
     * @param ip
     * @param port
     */
    public void registry(String service, String ip, int port) {
        //判断服务节点是否存在
        String path = ZK_ROOT_PATH + service + ZK_SPLIT + "provider";
        if (!exist(path)) {
            //创建永久节点
            create(path, false);
        }
        String instancePath = path + ZK_SPLIT + ip + ":" + port;
        create(instancePath, true);
    }

    /**
     * 创建节点
     *
     * @param path      创建节点路径
     * @param ephemeral 是否为临时节点
     */
    private void create(String path, boolean ephemeral) {
        try {
            if (!ephemeral) {
                //创建永久节点
                client.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath(path);
            } else {
                //创建临时节点
                client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath(path);
            }
        } catch (KeeperException.NodeExistsException e) {
            System.out.println("节点已存在，节点：" + path);
        } catch (Exception e) {
            e.printStackTrace();
            throw new RuntimeException(e.getMessage());
        }
    }

    /**
     * 校验节点是否存在
     *
     * @param path
     * @return
     */
    private boolean exist(String path) {
        try {
            Stat stat = client.checkExists().forPath(path);
            if (stat != null) {
                return true;
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return false;
    }

    /**
     * 发现服务（服务调用方）
     *
     * @param serviceName
     * @return
     */
    public List<String> discovery(String serviceName) {
        List<String> nodeList = LOAD_BALANCE_MAP.get(serviceName);
        //判断服务节点是否存在
        String path = ZK_ROOT_PATH + serviceName + ZK_SPLIT + "provider";
        try {
            //查找是否有子节点存在
            nodeList = client.getChildren().forPath(path);
            if (nodeList == null) {
                nodeList = new ArrayList<>();
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        //注册父节点监听器
        registryWatch(serviceName, path);
        return nodeList;
    }

    /**
     * 注册监视器
     *
     * @param serviceName
     * @param path
     */
    private void registryWatch(String serviceName, String path) {
        PathChildrenCache pathChildrenCache = new PathChildrenCache(client, path, true);
        try {
            pathChildrenCache.getListenable().addListener(new PathChildrenCacheListener() {
                @Override
                public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent pathChildrenCacheEvent) throws Exception {
                    System.out.println("监听到节点变更事件");
                    //遍历监听器列表对节点变更时间进行操作
                    listeners.stream().forEach(listener -> {
                        //通知消费者进行处理
                        listener.notify(serviceName, pathChildrenCacheEvent);
                    });
                }

            });
            // /lg-edu-rpc/com.lagou.edu.api.UserService/provider/127.0.0.1:8990
            /*
             * StartMode：初始化方式
             * POST_INITIALIZED_EVENT：异步初始化。初始化后会触发事件
             * NORMAL：异步初始化
             * BUILD_INITIAL_CACHE：同步初始化
             * */
            pathChildrenCache.start(PathChildrenCache.StartMode.BUILD_INITIAL_CACHE);
        } catch (Exception e) {
            e.printStackTrace();
        }


    }
}
